package de.mmis.core.pojop;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.lang.reflect.TypeVariable;
import java.net.Socket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.Hashtable;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.apache.log4j.Logger;

import de.mmis.core.base.Annotations;
import de.mmis.core.base.ProxyException;
import de.mmis.core.base.Toolbox;
import de.mmis.core.base.abstracttree.InnerNode;
import de.mmis.core.base.abstracttree.LeafNode;
import de.mmis.core.base.abstracttree.LeafNode.Encoding;
import de.mmis.core.base.abstracttree.Tree;
import de.mmis.core.base.abstracttree.Tree.Type;
import de.mmis.core.base.abstracttree.TreeParserException;
import de.mmis.core.base.abstracttree.serializer.DeserializableException;
import de.mmis.core.base.abstracttree.serializer.SerializableNetworkInvocationHandler;
import de.mmis.core.base.abstracttree.serializer.SerializerHelper;
import de.mmis.core.base.event.GenericObservable;
import de.mmis.core.base.event.Observable;
import de.mmis.core.base.event.Observer;
import de.mmis.core.base.gclass.GClass;
import de.mmis.core.base.gclass.GParameterized;
import de.mmis.core.sexpression.SExpressionParser;
import de.mmis.core.sexpression.SExpressionWriter;
import de.mmis.core.sexpression.UndoInputStream;

/**
 * Class providing static methods for creating and managing Interface Proxies
 * forwarding requests to a {@link NetworkPublisher}
 * 
 * @author Christian Koch
 * 
 */
public class NetworkProxy {

	private final static Logger LOGGER = Logger.getLogger(NetworkProxy.class);

	/** the {@link ThreadGroup} for all NetworkProxy threads */
	private static final ThreadGroup THREAD_GROUP = new ThreadGroup(
			NetworkProxy.class.getName());

	private NetworkProxy() {
		// hide constructor
	}

	/**
	 * creates new Proxy for specified interface, forwarding to
	 * {@link NetworkPublisher} on specified host and port, logging disabled
	 * 
	 * @param interfaze
	 *            interface to implement
	 * @param host
	 *            name or IP of NetworkPublisher server
	 * @param port
	 *            port number of NetworkPublisher server
	 * @return a Proxy implementing the specified interface that can be used
	 *         just as a local variable
	 * @throws UnknownHostException
	 *             specified host not found
	 * @throws IOException
	 *             error while connecting to NetworkPublisher server
	 */
	public static <T> T createProxy(ClassLoader classLoader,
			Class<T> interfaze, String host, int port)
			throws UnknownHostException, IOException {
		return createProxy(interfaze, classLoader, new Class[] { interfaze },
				host, port, true);
	}

	/**
	 * creates new Proxy for specified interface, forwarding to
	 * {@link NetworkPublisher} on specified host and port, logging disabled
	 * 
	 * @param id
	 *            unique identifier
	 * @param interfaces
	 *            interfaces to implement
	 * @param host
	 *            name or IP of NetworkPublisher server
	 * @param port
	 *            port number of NetworkPublisher server
	 * @return a Proxy implementing the specified interface that can be used
	 *         just as a local variable
	 * @throws UnknownHostException
	 *             specified host not found
	 * @throws IOException
	 *             error while connecting to NetworkPublisher server
	 */
	public static Proxy createProxy(String id, ClassLoader classLoader,
			Class<?>[] interfaces, String host, int port)
			throws UnknownHostException, IOException {
		return createProxy(Proxy.class, classLoader, interfaces, host, port,
				true);
	}

	/**
	 * creates new Proxy for specified interface, forwarding to
	 * {@link NetworkPublisher} on specified host and port
	 * 
	 * @param interfaze
	 *            interface to implement
	 * @param host
	 *            name or IP of NetworkPublisher server
	 * @param port
	 *            port number of NetworkPublisher server
	 * @return a Proxy implementing the specified interface that can be used
	 *         just as a local variable
	 * @throws UnknownHostException
	 *             specified host not found
	 * @throws IOException
	 *             error while connecting to NetworkPublisher server
	 */
	public static <T> T createProxy(Class<T> interfaze, String host, int port)
			throws UnknownHostException, IOException {
		return createProxy(interfaze, null, new Class[] { interfaze }, host,
				port, true);
	}

	/**
	 * creates new Proxy for specified interfaces, forwarding to
	 * {@link NetworkPublisher} on specified host and port
	 * 
	 * @param returnType
	 *            class specifying the return type
	 * @param interfazes
	 *            array of interfaces to implement
	 * @param host
	 *            name or IP of NetworkPublisher server
	 * @param port
	 *            port number of NetworkPublisher server
	 * @return a Proxy implementing the specified interface that can be used
	 *         just as a local variable
	 * @throws UnknownHostException
	 *             specified host not found
	 * @throws IOException
	 *             error while connecting to NetworkPublisher server
	 */
	@SuppressWarnings("unchecked")
	public static <T> T createProxy(Class<T> returnType,
			ClassLoader proxyClassLoader, Class<?>[] interfazes, String host,
			int port, boolean checkInterfaces) throws UnknownHostException,
			IOException {

		ClassLoader classLoader = proxyClassLoader;
		// determine class loader
		if (classLoader == null)
			classLoader = Toolbox.findCommonClassLoader(interfazes);
		if (classLoader == null) {
			LOGGER.warn("NetworkProxy: warning: cannot find suiting class loader, try default");
			classLoader = NetworkProxy.class.getClassLoader();
		}

		NetworkInvocationHandler invHandler = new NetworkInvocationHandler(
				classLoader, interfazes, host, port, checkInterfaces);

		for (Class<?> c : interfazes)
			invHandler.findObservableType(c);

		Thread thread = new Thread(THREAD_GROUP, invHandler,
				"invocation handler " + host + ":" + port);
		thread.setDaemon(true);
		thread.start();

		return (T) Proxy.newProxyInstance(classLoader, interfazes, invHandler);
	}

	/**
	 * closes connection to {@link NetworkPublisher} server
	 * 
	 * @param proxy
	 *            Proxy whose connection is closed
	 */
	public static void deleteProxy(Object proxy) {
		((NetworkInvocationHandler) Proxy.getInvocationHandler(proxy)).close();
	}

	public static Tree executeSingleCommand(String host, int port, Tree exp)
			throws IOException {
		Socket sock = new Socket(host, port);

		Tree ret = executeSingleCommand(sock, exp);

		SExpressionWriter.writeSExpression(new LeafNode(":quit"),
				sock.getOutputStream(), true, true);
		sock.close();

		return ret;
	}

	public static Tree executeSingleCommand(Socket sock, Tree exp)
			throws IOException {
		SExpressionWriter.writeSExpression(exp, sock.getOutputStream(), true,
				true);
		Tree ret;
		try {
			ret = SExpressionParser.parse(sock.getInputStream(), true);
		} catch (TreeParserException e) {
			throw new IOException(e);
		}
		return ret;
	}

	public static void executeSingleCommand(String host, int port, Tree exp,
			OutputStream out) throws IOException {
		Socket sock = new Socket(host, port);

		executeSingleCommand(sock, exp, out);

		SExpressionWriter.writeSExpression(new LeafNode(":quit"),
				sock.getOutputStream(), true, true);
		sock.close();
	}

	public static void executeSingleCommand(Socket sock, Tree exp,
			OutputStream out) throws IOException {
		SExpressionWriter.writeSExpression(exp, sock.getOutputStream(), true,
				true);
		try {
			SExpressionParser.parseAtom(
					new UndoInputStream(sock.getInputStream()), out);
		} catch (TreeParserException e) {
			throw new IOException(e);
		}
	}

	public static de.mmis.core.base.Type getDeviceType(String host, int port,
			ClassLoader cl) throws IOException, DeserializableException {
		return executeSingleCommand(host, port, new LeafNode(":type")).getAs(
				de.mmis.core.base.Type.class, cl);
	}

	public static de.mmis.core.base.Type getDeviceType(Socket sock,
			ClassLoader cl) throws IOException, DeserializableException {
		return executeSingleCommand(sock, new LeafNode(":type")).getAs(
				de.mmis.core.base.Type.class, cl);
	}

	public static Map<String, Object> getProperties(String host, int port)
			throws IOException, DeserializableException {
		return executeSingleCommand(host, port, new LeafNode(":properties"))
				.getAs(new GParameterized(Map.class, new GClass[] {
						GClass.fromClass(String.class),
						GClass.fromClass(Object.class) }));
	}

	public static Map<String, Object> getProperties(Socket sock)
			throws IOException, DeserializableException {
		return executeSingleCommand(sock, new LeafNode(":properties")).getAs(
				new GParameterized(Map.class, new GClass[] {
						GClass.fromClass(String.class),
						GClass.fromClass(Object.class) }));
	}

	public static de.mmis.core.base.Type[] getInterfaces(String host, int port)
			throws IOException, DeserializableException {
		return getInterfaces(host, port, NetworkProxy.class.getClassLoader());
	}

	public static de.mmis.core.base.Type[] getInterfaces(Socket sock)
			throws IOException, DeserializableException {
		return getInterfaces(sock, NetworkProxy.class.getClassLoader());
	}

	public static de.mmis.core.base.Type[] getInterfaces(String host, int port,
			ClassLoader cl) throws IOException, DeserializableException {
		return executeSingleCommand(host, port, new LeafNode(":interfaces"))
				.getAs(de.mmis.core.base.Type[].class, cl);
	}

	public static de.mmis.core.base.Type[] getInterfaces(Socket sock,
			ClassLoader cl) throws IOException, DeserializableException {
		return executeSingleCommand(sock, new LeafNode(":interfaces")).getAs(
				de.mmis.core.base.Type[].class, cl);
	}

	public static void getJarData(String host, int port, OutputStream out)
			throws IOException {
		executeSingleCommand(host, port, new LeafNode(":jar-data"), out);
	}

	public static void getJarData(Socket sock, OutputStream out)
			throws IOException {
		executeSingleCommand(sock, new LeafNode(":jar-data"), out);
	}

	/**
	 * class providing generic method invocation forwarding to a
	 * {@link NetworkPublisher}
	 * 
	 * @author Christian Koch
	 * 
	 */
	private static class NetworkInvocationHandler implements
			SerializableNetworkInvocationHandler, Runnable {
		private static final long DEFAULT_NWPROXY_TIMEOUT_MILLIS = 10000;
		private Socket sock;
		private InputStream in;
		private OutputStream out;
		private LinkedBlockingQueue<Tree> msgqueue = new LinkedBlockingQueue<Tree>();
		LinkedBlockingQueue<Tree> eventqueue = new LinkedBlockingQueue<Tree>();

		private Hashtable<TypeVariable<?>, GClass> typevarmap; // maps type
																// parameters of
																// parameterized
																// classes to
																// actual types
																// used (needed
																// because of
																// type erasure)

		GenericObservable genericObservable = null;

		private boolean keepListening;

		private String host;
		private int port;
		private Class<?>[] interfaces;

		private Thread eventListenThread;

		private ClassLoader classLoader;

		/**
		 * recursively process given type searching for implementation of
		 * Observable interface and creating the type parameter map
		 * 
		 * @param curtype
		 *            current type to process
		 */
		public void findObservableType(java.lang.reflect.Type curtype) {
			GClass.fromType(curtype, this.typevarmap);

			GClass[] obsparams = GClass.getGenericsOf(curtype,
					Observable.class, this.typevarmap);
			if (obsparams != null && obsparams.length == 1) {
				this.genericObservable = new GenericObservable(obsparams[0],
						this.classLoader);

				try {
					SExpressionWriter.writeSExpression(new LeafNode(":observe",
							Encoding.Token), this.out, true, true);
					/* SExpression sexp = */SExpressionParser.parse(this.in,
							true);
				} catch (IOException e) {
					LOGGER.warn("IOException while calling :observe", e);
				} catch (TreeParserException e) {
					LOGGER.warn("TreeParserException while calling :observe", e);
				}
			}
		}

		/**
		 * creates new invocation handler using given {@link NetworkPublisher}
		 * host and port
		 * 
		 * @param classLoader
		 * 
		 * @param host
		 *            host name or IP of NetworkPublisher
		 * @param port
		 *            port number of NetworkPublisher
		 * @throws UnknownHostException
		 *             specified host not found
		 * @throws IOException
		 *             error while connecting to NetworkPublisher server
		 */
		public NetworkInvocationHandler(ClassLoader classLoader,
				Class<?>[] interfaces, String host, int port,
				boolean checkInterfaces) throws UnknownHostException,
				IOException {
			this.typevarmap = new Hashtable<TypeVariable<?>, GClass>();

			this.classLoader = classLoader;
			this.interfaces = interfaces;

			this.host = host;
			this.port = port;

			this.sock = new Socket(host, port);

			if (checkInterfaces) {
				try {
					de.mmis.core.base.Type[] interfaceTypes = executeSingleCommand(
							sock, new LeafNode(":interfaces")).getAs(
							de.mmis.core.base.Type[].class, this.classLoader);

					nextinterface: for (Class<?> c : this.interfaces) {
						for (de.mmis.core.base.Type t : interfaceTypes)
							if (c.equals(t.getJavaClass()))
								continue nextinterface;
						throw new IllegalArgumentException("given interface "
								+ c
								+ " is not implemented by the remote object");
					}
				} catch (DeserializableException e) {
					LOGGER.error(
							"interfaces sent by NetworkPublisher were not valid",
							e);
					throw new IOException(
							"remote object interfaces cannot be checked", e);
				}
			}

			this.in = this.sock.getInputStream();
			this.out = this.sock.getOutputStream();

			this.keepListening = true;

			eventListenThread = new Thread(THREAD_GROUP, new Runnable() {
				@Override
				public void run() {
					while (keepListening) {
						Tree event;
						try {
							event = NetworkInvocationHandler.this.eventqueue
									.take();
							NetworkInvocationHandler.this.genericObservable
									.fireEvent(event);
						} catch (InterruptedException e) {
							if (keepListening)
								LOGGER.error("take unexpectedly interrupted", e);
							else
								return;
						}
					}

				}

			});
			eventListenThread.setName("eventListenThread " + host + ":" + port);
			eventListenThread.setDaemon(true);
			eventListenThread.start();
		}

		/**
		 * forward invocation of Proxy method to {@link NetworkPublisher}
		 * 
		 * @param proxy
		 *            Proxy object invocation was made on
		 * @param method
		 *            method invoked
		 * @param args
		 *            array of method parameters
		 * @return result of method call
		 * @throws ProxyException
		 *             if NetworkPublisher returned error message
		 * @throws Throwable
		 *             any other error that may occur
		 */

		@Override
		public Object invoke(Object proxy, Method method, Object[] args)
				throws Throwable {

			if (this.genericObservable != null) {
				if (method.getName().equals("addObserver") && args.length == 1) {
					this.genericObservable.addObserver((Observer<?>) args[0]);
					return null;
				} else if (method.getName().equals("removeObserver")
						&& args.length == 1) {
					this.genericObservable
							.removeObserver((Observer<?>) args[0]);
					return null;
				}
			}

			InnerNode funcCall = new InnerNode();
			funcCall.add(new LeafNode(method.getName()));

			if (args != null) {
				for (int i = 0; i < args.length; i++)
					funcCall.add(Tree.fromObject(args[i]));
			}
			Tree retexp;
			synchronized (this.msgqueue) { // when multiple method invocations
											// occur at the same time, it
											// wouldn't be ensured that every
											// call gets its own return value
											// back. NetworkPublisher process
											// method calls one by one.
				SExpressionWriter.writeSExpression(funcCall, this.out, true,
						true);

				long timeout = Annotations.getAnnotationTimeout(method, args,
						DEFAULT_NWPROXY_TIMEOUT_MILLIS);

				retexp = this.msgqueue.poll(timeout, TimeUnit.MILLISECONDS);
				if (retexp == null)
					throw new TimeoutException(
							"timeout while waiting for answer from NetworkPublisher");
			}

			if (retexp.getType() == Type.INNER
					&& ((InnerNode) retexp).getNumberOfSubTrees() > 0
					&& ((InnerNode) retexp).getSubTree(0).getType() == Type.LEAF
					&& ((LeafNode) ((InnerNode) retexp).getSubTree(0))
							.equalsUTF8String(":error")) {
				throw ProxyException.generateException(retexp, classLoader);
			}

			if (method.getReturnType() == void.class)
				return null;
			Object retval = SerializerHelper.deserializeObject(retexp, GClass
					.fromType(method.getGenericReturnType(), this.typevarmap),
					this.classLoader, new Hashtable<String, Object>());

			return retval;
		}

		@Override
		public void finalize() {
			close();
		}

		/**
		 * closes {@link NetworkPublisher} socket
		 */
		public void close() {
			this.keepListening = false;

			try {
				eventListenThread.interrupt();
			} catch (SecurityException e) {
				LOGGER.warn("error interrupting event listen thread", e);
			}

			try {
				if (this.out != null) {
					SExpressionWriter.writeSExpression(new LeafNode(":quit",
							Encoding.Token), this.out, true, true);
				}
			} catch (IOException e) {
				LOGGER.warn("error sending \":quit\" message", e);
			}
			this.out = null;
			try {
				if (this.sock != null)
					this.sock.close();
			} catch (IOException e) {
				// ignore failures
			}
			this.sock = null;
		}

		@Override
		public void run() {
			try {
				while (keepListening && this.sock != null
						&& !this.sock.isClosed()) {
					// System.err.println("waiting for event ... "+Thread.currentThread().getName()+"="+Thread.currentThread().getId());
					Tree exp = SExpressionParser.parse(this.in, true);
					// System.err.println("got event "+exp+" / "+Thread.currentThread().getName()+"="+Thread.currentThread().getId());

					if (exp.getType() == Type.INNER
							&& ((InnerNode) exp).getNumberOfSubTrees() == 2
							&& ((InnerNode) exp).getSubTree(0).getType() == Type.LEAF
							&& ((LeafNode) ((InnerNode) exp).getSubTree(0))
									.equalsUTF8String(":event")) {
						Tree event = ((InnerNode) exp).getSubTree(1);
						// genericObservable.fireEvent(event);
						this.eventqueue.add(event);
					} else
						this.msgqueue.add(exp);
				}
			} catch (TreeParserException e) { // normal socket closing if
												// keepListening is
												// false
				if (this.keepListening)
					LOGGER.error(e, e);
			} catch (SocketException e) {
				if (this.keepListening)
					LOGGER.error(e, e);
			} catch (IOException e) {
				if (this.keepListening)
					LOGGER.error(e, e);
			}

		}

		@Override
		public String getHost() {
			return this.host;
		}

		@Override
		public Class<?>[] getImplementingInterfaces() {
			return this.interfaces;
		}

		@Override
		public int getPort() {
			return this.port;
		}

		@Override
		public Class<?> getProxyCreationClass() {
			return NetworkProxy.class;
		}

		@Override
		public String getID() {
			return null;
		}
	}
}
